/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */
package users;

import java.util.UUID;

import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.TopicSubscriber;

import org.apache.qpid.AMQException;
import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.AMQTopic;
import org.apache.qpid.client.TopicPublisherAdapter;
import org.apache.qpid.client.message.AbstractJMSMessage;
import org.apache.qpid.client.message.JMSTextMessage;
import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.url.URLSyntaxException;

/**
 * Data needed by a user to connect with a Qpid broker and interact with it.
 */
public class UserConnection {

    /** User to whom the connection belongs. */
    private User user;
    /**
     * Getter for the variable 'user'.
     * @return variable 'user'
     */
    public User getUser() {
        return user;
    }

    /** Qpid broker. */
    private final String broker;
    /** Port of the Qpid broker. */
    private final int port;
    /** Default AMQP port. */
    private static final int DEFAULT_PORT = 5672;
    /** Name of the virtual host to use. */
    private static final String VIRTUAL_HOST = "news";
    /** Name of the exchange to use. */
    private final String exchange;
    /** Type of the exchange to use. */
    private static final String EXCHANGE_TYPE = "topic";

    /**
     * Getter for the variable 'broker'.
     * @return variable 'broker'
     */
    public String getBroker() {
        return broker;
    }
    /**
     * Getter for the variable 'port'.
     * @return variable 'port'
     */
    public int getPort() {
        return port;
    }
    /**
     * Getter for the variable 'exchange'.
     * @return variable 'exchange'
     */
    public String getExchange() {
        return exchange;
    }

    /** Connection with the broker. */
    private AMQConnection connection;
    /** Session in the connection with the broker. */
    private AMQSession session;

    /**
     * Getter for the variable 'connection'.
     * @return variable 'connection'
     */
    public AMQConnection getConnection() {
        return connection;
    }
    /**
     * Setter for the variable 'connection'.
     * @param connection  new value of the variable 'connection'
     */
    public void setConnection(final AMQConnection connection) {
        this.connection = connection;
    }
    /**
     * Getter for the variable 'session'.
     * @return variable 'session'
     */
    public AMQSession getSession() {
        return session;
    }
    /**
     * Setter for the variable 'session'.
     * @param session  new value of the variable 'session'
     */
    public void setSession(final AMQSession session) {
        this.session = session;
    }


    /**
     * Constructor.
     *
     * @param user  to whom the connection belongs
     * @param broker  to connect to
     * @param port  in which the broker is listening
     * @param exchange  to use in the broker
     */
    public UserConnection(final User user, final String broker, final int port,
                          final String exchange) {
        this.user = user;
        this.broker = broker;
        this.port = port;
        this.exchange = exchange;
    }

    /**
     * Constructor using the default port.
     *
     * @param user  to whom the connection belongs
     * @param broker  to connect to
     * @param exchange  to use in the broker
     */
    public UserConnection(final User user, final String broker,
                          final String exchange) {
        this.user = user;
        this.broker = broker;
        this.port = DEFAULT_PORT;
        this.exchange = exchange;
    }

    /**
     * Create and start a connection to the broker.
     * @throws AMQException  if there is any problem
     * @throws URLSyntaxException  if there is any problem
     * @throws JMSException  if there is any problem
     */
    protected void openConn()
            throws URLSyntaxException, AMQException, JMSException {

        String connURL = "amqp://" + getUser().getUsername() + ":"
                + getUser().getPassword() + "@" + getUser().getClientId() + "/"
                + VIRTUAL_HOST + "?brokerlist='tcp://" + getBroker() + ":"
                + getPort() + "'";

        //create a connection
        setConnection(new AMQConnection(connURL));
        System.out.println(getUser().getClientId() + " created connection to "
                           + getBroker() + " at port " + getPort());
        //now the connection is setup up start it
        getConnection().start();

    }

    /**
     * Close the connection.
     *
     * @throws JMSException  if there is any problem
     */
    public void closeConn() throws JMSException {
        getConnection().close();
    }

    /**
     * Create a session, non transacted and auto acknowledged, in the
     * connection with the broker.
     *
     * @return  AMQSession created
     * @throws JMSException  if there is any problem
     */
    protected AMQSession createSession() throws JMSException {
        return (AMQSession) getConnection().createSession(
                false, AMQSession.AUTO_ACKNOWLEDGE);
//      System.out.println(getUser().getName() + " created topic session");
    }

    /**
     * Declare the exchange. It will be created if it doesn't exist.
     *
     * @throws AMQException  if there is any problem
     */
    protected void declareExchange() throws AMQException {
        AMQShortString exchName = new AMQShortString(getExchange());
        AMQShortString exchType = new AMQShortString(EXCHANGE_TYPE);
        getSession().declareExchange(exchName, exchType, true);
//         System.out.println("Exchange " + getEXCHANGE() + " declared in "
//                            + getBROKER());
    }

    /**
     * Get the connection ready.
     *
     * @throws JMSException  if there is any problem
     * @throws AMQException  if there is any problem
     * @throws URLSyntaxException  if there is any problem
     */
    public void initialize()
            throws JMSException, AMQException, URLSyntaxException {
        openConn();
        setSession(createSession());
        declareExchange();
    }

    /**
     * Create a publisher to the given topic.
     *
     * @param topicName  the topic to create a publisher to
     * @return  a publisher to the given topic
     * @throws JMSException  if there are problems
     */
    protected TopicPublisherAdapter getPublisherForTopic(
            final String topicName) throws JMSException {
        //create topicURL
        String tURL = "topic://" + getExchange() + "/?routingkey='" + topicName
                      + "'&exclusive='false'&autodelete='true'";
        //create topic
        AMQTopic topic = (AMQTopic) getSession().createTopic(tURL);

        //return publisher
        return (TopicPublisherAdapter) getSession().createPublisher(topic);
    }

    /**
     * Create a subscriber to the given topic.
     *
     * @param topicName  topic to create the subscriber to
     * @return  a subscriber to the given topic
     * @throws JMSException  if there are problems
     */
    protected TopicSubscriber getSubscriberForTopic(final String topicName)
            throws JMSException {
        //create topicURL
        String tURL = "topic://" + getExchange() + "/?routingkey='" + topicName
                      + "'&exclusive='false'&autodelete='true'";
        //create topic
        AMQTopic topic = (AMQTopic) getSession().createTopic(tURL);

        //get a random subscription ID
        String subscriptionId = UUID.randomUUID().toString();
        //return a subscriber
        return getSession().createDurableSubscriber(topic, subscriptionId);
    }

    /**
     * Create a text message with the given text and without properties and
     * makes the given publisher send it.
     *
     * @param publisher  the publisher who will send the message
     * @param text  text of the message
     * @throws JMSException  if there are problems
     */
    protected void sendMessage(final TopicPublisherAdapter publisher,
                               final String text) throws JMSException {
        //create the message to send
        JMSTextMessage txtMessage
                = (JMSTextMessage) getSession().createTextMessage(text);
        //send the message
        publisher.publish(txtMessage, DeliveryMode.PERSISTENT,
                          AbstractJMSMessage.DEFAULT_PRIORITY,
                          AbstractJMSMessage.DEFAULT_TIME_TO_LIVE);
    }

    /**
     * Checks if the connection is closed or closing.
     *
     * @return  true if the connection is not closed or closing
     */
    protected boolean isConnected() {
        //TODO: decide if we should take into account 'isFailingOver'
        return (!getConnection().isClosed()) && (!getConnection().isClosing()
                && (!getSession().isClosed()) && (!getSession().isClosing()));
    }

}
